package ar.edu.itba.pod.legajo48240.event.threads;

import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.BlockingQueue;

import org.apache.log4j.Logger;

import ar.edu.itba.event.EventInformation;
import ar.edu.itba.event.RemoteEventDispatcher;
import ar.edu.itba.node.NodeInformation;
import ar.edu.itba.pod.legajo48240.event.LocalEventDispatcher;
import ar.edu.itba.pod.legajo48240.model.Host;
import ar.edu.itba.pod.legajo48240.simulation.DistributedSimulationManager;
import ar.edu.itba.pod.legajo48240.utils.Net;
import ar.edu.itba.pod.legajo48240.utils.Utils;
import ar.edu.itba.pod.thread.CleanableThread;

public class EventTakerThread extends CleanableThread {

	/**
	 * new events to publish
	 */

	private final BlockingQueue<EventInformation> incoming;

	/**
	 * already broadcasted events
	 */

	private final BlockingQueue<EventInformation> processed;

	private final Host localhost;
	
	private final Random RANDOM= new Random();

	private final static Logger LOGGER = Logger
			.getLogger(EventTakerThread.class);

	public EventTakerThread(String name,
			BlockingQueue<EventInformation> incoming,
			BlockingQueue<EventInformation> processed, Host localhost) {
		super(name);
		this.incoming = incoming;
		this.processed = processed;
		this.localhost = localhost;
	}

	@Override
	public void run() {
		super.run();

		try {
			while (!shouldFinish()) {
				EventInformation eventInformation = incoming.take();
				processed.put(eventInformation);

				// broadcast to local agents

				((LocalEventDispatcher) (((DistributedSimulationManager) localhost
						.simulation()).dispatcher())).localPublish(
						eventInformation.source(), eventInformation.event());

				// broadcast to other host network
				
				remotePublish(eventInformation);

			}
		} catch (InterruptedException e) {
			LOGGER.debug("finishing queue events thread...");
		}
	}
	
	
	
	public void remotePublish(EventInformation eventInformation) throws InterruptedException{
		
		Set<NodeInformation> nodes;
		try {
			nodes = localhost.network().connectedNodes();
			
			if(!nodes.isEmpty()){
				
				Set<NodeInformation> copy = Utils.clone(nodes);
				
				double total=copy.size();
				double success = total ;

				for(NodeInformation node:copy){
					if( !node.equals(localhost.node()) && !node.equals(eventInformation.nodeId()) && RANDOM.nextDouble() < success/total ){
						try {
							RemoteEventDispatcher dispatcher=Net.getDispatcher(node.host(),node.port());
							if(!dispatcher.publish(eventInformation)){
								/** stop broadcast for known events */
								success--;
							}else{
//								LOGGER.debug(localhost.id() + " pulished in " + node.id() + " event: " + eventInformation.event().getClass().getSimpleName());
							}
						} catch (NotBoundException e) {
							LOGGER.warn("Disconnect " + node.host() + " from " + localhost.host() +" because connection was broken" );
							try {
								localhost.network().disconnectFromGroup(node);
							} catch (NotBoundException ignore) {} 
						}
					}
				}
			}
		} catch (RemoteException ignore) { } // connectedNodes as local method	

	}

}
